package com.lg.monitor

import java.sql.Connection

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter


/**
  * 写入mysql的writer
  */
object MysqlWriter {

  // 载入mysql驱动
  val JDBC_DRIVER = "com.mysql.jdbc.Driver"
  val DB_URL = "jdbc:mysql://localhost:3306/test"
  val USER = "root"
  val PASS = "123456"
  // 注册 JDBC 驱动
  Class.forName(JDBC_DRIVER)


  //获取jdbc连接
  def getConnection:Connection = {
    //todo  返回一个连接
    import java.sql.DriverManager
    DriverManager.getConnection(DB_URL, USER, PASS)
  }
}
class MysqlWriter extends ForeachWriter[BusInfo] {
  var conn: Connection = _

  val sql = "insert into real_time(deploy_num,lglat) values(?,?)"

  //开启连接
  override def open(partitionId: Long, epochId: Long): Boolean = {
    conn = MysqlWriter.getConnection
    true
  }

  //处理数据
  override def process(value: BusInfo): Unit = {
    //把数据写入mysql , 插入一张real_time表
    val deployNum = value.deployNum
    val lglat: String = value.lglat
    val pstm = conn.prepareStatement(sql)
    pstm.setString(1, deployNum)
    pstm.setString(2, lglat)
    pstm.executeUpdate
  }

  //释放连接
  override def close(errorOrNull: Throwable): Unit = {
    conn.close()
  }
}
